package com.mq.consumer.simple.v2;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @Classname 消费者组测试
 * @Description
 * @Version 1.0.0
 * @Date 2025/3/28 16:51
 * @Created by ncs
 */
public class CustomConsumerGroup {

    static String kafkaServer = "192.168.125.100:9093,192.168.125.100:9094,192.168.125.100:9095";

    public static void main(String[] args) {
        int threadSize = 3;
        for (int i = 0; i < threadSize; i++) {
            new Thread(() -> {
                String name = Thread.currentThread().getName();
                System.out.println("线程=" + name + "准备就绪");
                Properties properties = new Properties();
                properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
                properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
                properties.put(ConsumerConfig.GROUP_ID_CONFIG, "first-consumer-group");
                KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
                kafkaConsumer.subscribe(Collections.singleton("first-topic"));
                while (true){
                    ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(20));
                    consumerRecords.forEach(n->{
                        System.out.println("线程=" + name + "消费了数据\t" + n);
                    });
                }
            }).start();
        }
    }
}
